#pragma once
#include <string>
#include <iostream>
#include <memory>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

#include "consumer.hpp"
#include "../mqcommom/threadpool.hpp"
#include "../mqcommom/helper.hpp"
#include "../mqcommom/proto.pb.h"
#include "../mqcommom/msg.pb.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"

/*
* 信道模块模块
主要负责请求的发送，服务端需要什么这里就设计什么
声明删除交换机/声明删除队列/绑定解绑队列/生产消费/waitResponse接收服务的响应，等待确认
*成员 consumer mutex condition_varible map<rid,response> cid 一个信道对应一个消费者 conn(通信连接) codec(协议分派器)
*注意：每一个请求都有自己的响应

*/
using namespace ns_helper;
namespace ns_rabbitmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    class Channel
    {
        using BasicConsumerResponseptr = std::shared_ptr<BasicConsumerResponse>;
        using BasicCommResponseptr = std::shared_ptr<BasicCommResponse>;

    public:
        Channel(const muduo::net::TcpConnectionPtr &conn, const ProtobufCodecPtr &codec)
            : _cid(UUIDHelp::uuid()), _conn(conn), _codec(codec)
        {
        }
        ~Channel()
        {
            BasicCancel();
        }
        std::string cid()
        {
            return _cid;
        }
        bool OpenChannel()
        {
            std::string rid = UUIDHelp::uuid();
            OpenChannelRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            _codec->send(_conn, req);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            // 返回
            return resp->ok();
        }

        void CloseChannel()
        {
            std::string rid = UUIDHelp::uuid();
            CloseChannelRequest req;
            req.set_cid(_cid);
            req.set_rid(rid);
            _codec->send(_conn, req);
            // 阻塞等待回应
            WaitResponse(rid);
            // 返回
            return;
        }
        bool DeclareExchange(const std::string &name, ExchangeType exchangetype,
                             bool durable, bool autodelete, google::protobuf::Map<std::string, std::string> args)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            DeclearExchangeRequest e_reque;
            e_reque.set_rid(rid);
            e_reque.set_cid(_cid);
            e_reque.set_exchange_name(name);
            e_reque.set_exchange_type(exchangetype);
            e_reque.set_durable(durable);
            e_reque.set_auto_delect(autodelete);
            e_reque.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_conn, e_reque);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            // 返回
            return resp->ok();
        }

        void DeleteExchange(const std::string &name)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            DeleteExchangeRequest e_reque;
            e_reque.set_rid(rid);
            e_reque.set_cid(_cid);
            e_reque.set_exchange_name(name);
            // 发送请求
            _codec->send(_conn, e_reque);
            // 阻塞等待回应
           WaitResponse(rid);
            return;
        }

        bool DeclareQueue(const std::string &name, bool exclusive,
                          bool durable, bool auto_delete, google::protobuf::Map<std::string, std::string> args)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            DeclearQueueRequest q_reque;
            q_reque.set_rid(rid);
            q_reque.set_cid(_cid);
            q_reque.set_queue_name(name);
            q_reque.set_exclusive(exclusive);
            q_reque.set_durable(durable);
            q_reque.set_auto_delect(auto_delete);
            q_reque.mutable_args()->swap(args);
            // 发送请求
            _codec->send(_conn, q_reque);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            // 返回
            return resp->ok();
        }
        void DeleteQueue(const std::string &name)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            DeleteQueueRequest q_reque;
            q_reque.set_rid(rid);
            q_reque.set_cid(_cid);
            q_reque.set_queue_name(name);
            // 发送请求
            _codec->send(_conn, q_reque);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            return;
        }

        // 绑定和解绑
        bool Bind(const std::string &exchange_name, const std::string &queue_name, const std::string &binding_key, bool durable)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            QueueBindRequest bind_req;
            bind_req.set_rid(rid);
            bind_req.set_cid(_cid);
            bind_req.set_queue_name(queue_name);
            bind_req.set_exchange_name(exchange_name);
            bind_req.set_durable(durable);
            bind_req.set_bindkey(binding_key);
            // 发送请求
            _codec->send(_conn, bind_req);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            // 返回
            return resp->ok();
        }
        void UnBind(const std::string &exchange_name, const std::string &queue_name)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            QueueUnbindRequest bind_req;
            bind_req.set_rid(rid);
            bind_req.set_cid(_cid);
            bind_req.set_queue_name(queue_name);
            bind_req.set_exchange_name(exchange_name);
            // 发送请求
            _codec->send(_conn, bind_req);
            // 阻塞等待回应
            WaitResponse(rid);
            // 返回
            return;
        }

        bool BasicPublish(const std::string &ename, BasicProperties *bp, const std::string &body)
        {
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            BasicPublishRequest bind_req;
            bind_req.set_rid(rid);
            bind_req.set_cid(_cid);
            bind_req.set_exchange_name(ename);
            if (bp)
            {
                bind_req.mutable_propertie()->set_id(bp->id());
                bind_req.mutable_propertie()->set_routing_key(bp->routing_key());
                bind_req.mutable_propertie()->set_deliver_mode(bp->deliver_mode());
            }
            bind_req.set_body(body);
            std::cout<<"req_routing_key:"<<bind_req.propertie().routing_key()<<std::endl;
            // 发送请求
            _codec->send(_conn, bind_req);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            // 返回
            return resp->ok();
        }

        void BasicAck(const std::string &msg_id)
        {
            if (!_consumer.get())
            {
                return;
            }
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            BasicAckRequest ack_req;
            ack_req.set_rid(rid);
            ack_req.set_cid(_cid);
            ack_req.set_queue_name(_consumer->_qname);
            ack_req.set_msg_id(msg_id);
            // 发送请求
            _codec->send(_conn, ack_req);
            // 阻塞等待回应
            WaitResponse(rid);
            // 返回
            return;
        }

        void BasicCancel()
        {
            if (!_consumer.get())
            {
                return;
            }
            // 构建请求
            std::string rid = UUIDHelp::uuid();
            BasicCancelRequest cancel_req;
            cancel_req.set_rid(rid);
            cancel_req.set_cid(_cid);
            cancel_req.set_consumer_tag(_consumer->_tag);
            cancel_req.set_qname(_consumer->_qname);

            // 发送请求
            _codec->send(_conn, cancel_req);
            // 阻塞等待回应
            WaitResponse(rid);
            // 返回
            _consumer.reset();
            return;
        }

        //订阅消息
        bool BasicConsume(const std::string &tag, const std::string &qname, bool auto_delete, ConsumCallback cb)
        {
            if (_consumer.get())
            {
                ELOG("信道%s,已经存在消费者%s!创建失败", _cid.c_str(), _consumer->_tag.c_str());
                return false;
            }
            std::string rid = UUIDHelp::uuid();
            BasicConsumerRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_qname(qname);
            req.set_consumer_tag(tag);
            req.set_auto_ack(auto_delete);
            _codec->send(_conn, req);
            // 阻塞等待回应
            BasicCommResponseptr resp = WaitResponse(rid);
            if (!resp->ok())
            {
                ELOG( "添加订阅失败!");
                return false;
            }
            _consumer = std::make_shared<Consumer>(tag, qname, auto_delete, cb);
            return true;
        }

        void PutBasicResponse(const BasicCommResponseptr &resp)
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            // 往map中方响应
            _response.insert(std::make_pair(resp->rid(), resp));
            // 唤醒所有的阻塞
            _cv.notify_all();
        }

        void Consume(const BasicConsumerResponseptr &resp)
        {
            if (!_consumer.get())
            {
                ELOG("找不到指定消费者!");
                return;
            }
            if (resp->consumer_tag() != _consumer->_tag)
            {
                ELOG( "标签与指定消费者不一致!");
                return;
            }
            _consumer->_cb(resp->consumer_tag(), resp->mutable_propertie(), resp->body());
        }

    private:
        BasicCommResponseptr WaitResponse(const std::string &rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cv.wait(lock, [this, &rid]()
                     { return _response.find(rid) != _response.end(); });
            BasicCommResponseptr resp = _response[rid];
            _response.erase(rid);
            return resp;
        }

    private:
        std::string _cid;
        ProtobufCodecPtr _codec;
        muduo::net::TcpConnectionPtr _conn;
        ConsumerPtr _consumer;
        std::mutex _mutex;
        std::condition_variable _cv;
        std::unordered_map<std::string, BasicCommResponseptr> _response;
    };
    using ChannelPtr = std::shared_ptr<Channel>;
    /*管理模块 对外提供信道的增删查
     用map管理一个连接的多个信道 以信道id为key
     mutex保护
    */
    class ChannelManager
    {
    public:
        ChannelManager() {}
        ChannelPtr CreateChannel(const muduo::net::TcpConnectionPtr &conn,
                                 const ProtobufCodecPtr &codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ChannelPtr channel = std::make_shared<Channel>(conn, codec);
            _channels.insert(std::make_pair(channel->cid(), channel));
            return channel;
        }

        void RemoveChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                ELOG( "不存在指定%s的信道，删除失败", cid.c_str());
                return;
            }
            _channels.erase(it);
        }

        ChannelPtr GetChannel(const std::string &cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                ELOG("不存在指定%s的信道.获取失败", cid.c_str());
                return ChannelPtr();
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, ChannelPtr> _channels;
    };
    using ChannelManagerPtr = std::shared_ptr<ChannelManager>;
}